package org.example.redis;

import org.apache.commons.lang3.RandomStringUtils;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import redis.clients.jedis.Jedis;

public class RedisSource implements SourceFunction<String> {
    private volatile boolean isRunning = true;
    private final String redisHost;
    private final int redisPort;
    private final String redisKey;

    public RedisSource(String redisHost, int redisPort, String redisKey) {
        this.redisHost = redisHost;
        this.redisPort = redisPort;
        this.redisKey = redisKey;
        new Thread(()->{
            Jedis jedis = new Jedis(redisHost, redisPort);
            while (true){
                String str = RandomStringUtils.randomAscii(5);
                System.out.println(str);
                jedis.lpush(redisKey,str);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).start();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        try (Jedis jedis = new Jedis(redisHost, redisPort)) {
            while (isRunning) {
                String value = jedis.lpop(redisKey);
                if (value != null) {
                    ctx.collect(value);
                }
                Thread.sleep(1000); // 每秒拉取一次数据
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
